package com.sijibao.gather.consumer.kafka;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaConsumerAdapter
{
	private static final Logger logger = LoggerFactory
			.getLogger(KafkaConsumerAdapter.class);
	private KafkaConsumerConfig kafkaConsumerConfig;

	public void setKafkaConsumerConfig(KafkaConsumerConfig kafkaConsumerConfig)
	{
		this.kafkaConsumerConfig = kafkaConsumerConfig;
	}

	private KafkaConsumer<String, String> consumer;

	public void init()
	{
		Map<String, Object> config = new HashMap<String, Object>();
		config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
				kafkaConsumerConfig.getBootstrapServers());
		config.put(ConsumerConfig.CLIENT_ID_CONFIG,
				kafkaConsumerConfig.getClientId());
		config.put(ConsumerConfig.GROUP_ID_CONFIG,
				kafkaConsumerConfig.getGroupId());
		config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
				kafkaConsumerConfig.isEnableAutoCommit());
		config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,
				kafkaConsumerConfig.getAutoCommitIntervalMs());
		config.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,
				kafkaConsumerConfig.getSessionTimeoutMs());

		consumer = new KafkaConsumer<String, String>(config,
				new StringDeserializer(), new StringDeserializer());
		consumer.subscribe(
				Collections.singletonList(kafkaConsumerConfig.getTopic()));
	}

	public KafkaConsumer<String, String> getConsumer()
	{
		return consumer;
	}

	public void destroy()
	{
		logger.info("Consumer KafkaAdapter Begin Destroy...");
		if (consumer != null)
		{
			consumer.close();
			consumer = null;
		}
		logger.info("Consumer KafkaAdapter Destroy Success");
	}

}
